#include "SessionManager.h"
#include "TCPSession.h"
#include "ServerApp.h"

SessionManager::SessionManager()
	: m_work(m_io)
{
	m_SocketIDGen = 0;
}

SessionManager::~SessionManager()
{

}

void SessionManager::Start(int threadCount, 
	std::vector<boost::asio::ip::tcp::endpoint> & ips, 
	std::vector<boost::asio::ip::tcp::endpoint> & https)
{
	m_pLogger = GetServerApp().getLoggerManager()->getLogger("system");

	m_io_sockets.init(threadCount);

	for (size_t n = 0; n < ips.size(); ++n)
	{
		startTcp(ips[n]);
	}

	for (size_t n = 0; n < https.size(); ++n)
	{
		startHttp(https[n]);
	}


	m_thread.reset(new boost::thread( boost::bind(&boost::asio::io_service::run,& m_io)));
}

void SessionManager::StopAll()
{
	m_io.dispatch(boost::bind(&SessionManager::RealStop, this));
	m_thread->join();
}

void SessionManager::startTcp(const boost::asio::ip::tcp::endpoint & ep)
{
	boost::shared_ptr<boost::asio::ip::tcp::acceptor> apt(
		new boost::asio::ip::tcp::acceptor(m_io));

	try
	{
		apt->open(ep.protocol());
		apt->set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
		apt->bind(ep);
		apt->listen();
	}
	catch (const std::exception & e)
	{
		m_pLogger->Write(LOG_LEVEL_FATAL,"[SessionManager::startTcp] start address{%s:%u} failed err=%s\r\n",
			ep.address().to_string().c_str(), (uint32_t)ep.port(), e.what());
		abort();
		return;
	}

	m_pLogger->Write(LOG_LEVEL_DEBUG, "[SessionManager::startTcp] start address{%s:%u} success \r\n",
		ep.address().to_string().c_str(), (uint32_t)ep.port());

	m_acceptors.push_back(apt);

	boost::shared_ptr<boost::asio::ip::tcp::socket> pSocket(
		new boost::asio::ip::tcp::socket(get_io_service()));

	apt->async_accept(*pSocket,
		boost::bind(&SessionManager::OnTcpAccept, this,
		boost::asio::placeholders::error,
		pSocket, apt));
}

void SessionManager::startHttp(const boost::asio::ip::tcp::endpoint & ep)
{
	boost::shared_ptr<boost::asio::ip::tcp::acceptor> apt(
		new boost::asio::ip::tcp::acceptor(m_io));

	try
	{
		apt->open(ep.protocol());
		apt->set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
		apt->bind(ep);
		apt->listen();
	}
	catch (const std::exception & e)
	{
		m_pLogger->Write(LOG_LEVEL_ERROR,"[SessionManager::startHttp] start address{%s:%u} failed err=%s\r\n",
			ep.address().to_string().c_str(), (uint32_t)ep.port(), e.what());
		abort();
		return;	
	}

	m_pLogger->Write(LOG_LEVEL_DEBUG, "[SessionManager::startHttp] start address{%s:%u} success \r\n",
		ep.address().to_string().c_str(), (uint32_t)ep.port());

	m_acceptors.push_back(apt);

	boost::shared_ptr<boost::asio::ip::tcp::socket> pSocket(
		new boost::asio::ip::tcp::socket(get_io_service()));

	apt->async_accept(*pSocket,
		boost::bind(&SessionManager::OnHttpAccept, this,
		boost::asio::placeholders::error,
		pSocket, apt));
}

void SessionManager::OnTcpAccept(const boost::system::error_code & e, 
	boost::shared_ptr<boost::asio::ip::tcp::socket> pSocket, 
	boost::shared_ptr<boost::asio::ip::tcp::acceptor> pAcceptor)
{

	if (m_SessionMap.size() < 10000)
	{
		SocketID id = m_SocketIDGen++;
		boost::shared_ptr<TCPSession> pSession(new TCPSession(pSocket, *this, id, Session_TCP));
		try
		{
			pSession->Start();		
			m_SessionMap.insert(std::make_pair(id, pSession));
		}
		catch (const std::exception & e)
		{
			boost::asio::ip::tcp::endpoint ep;
			try
			{
				ep = pSocket->remote_endpoint();
			}
			catch (...)
			{
			}
			
			m_pLogger->Write(LOG_LEVEL_INFO, "[SessionManager::OnTcpAccept] accept from {%s:%u}, error : %s \r\n",
				ep.address().to_string().c_str(), ep.port(), e.what());

			try
			{
				pSocket->shutdown(boost::asio::ip::tcp::socket::shutdown_both);
				pSocket->close();
			}
			catch(...) {}
		}
	}	
	else
	{
		try
		{
			pSocket->shutdown(boost::asio::ip::tcp::socket::shutdown_both);
			pSocket->close();
		}
		catch (...) {}
	}

	boost::shared_ptr<boost::asio::ip::tcp::socket> pNewSocket(
		new boost::asio::ip::tcp::socket(get_io_service()));

	pAcceptor->async_accept(*pNewSocket,
		boost::bind(&SessionManager::OnTcpAccept, this,
		boost::asio::placeholders::error,
		pNewSocket, pAcceptor));
}


void SessionManager::OnHttpAccept(const boost::system::error_code & e,
	boost::shared_ptr<boost::asio::ip::tcp::socket> pSocket,
	boost::shared_ptr<boost::asio::ip::tcp::acceptor> pAcceptor)
{
	SocketID id = m_SocketIDGen++;
	boost::shared_ptr<TCPSession> pSession(new TCPSession(pSocket, *this, id, Session_HTTP));
	pSession->Start();

	m_SessionMap.insert(std::make_pair(id, pSession));

	boost::shared_ptr<boost::asio::ip::tcp::socket> pNewSocket(
		new boost::asio::ip::tcp::socket(get_io_service()));

	pAcceptor->async_accept(*pNewSocket,
		boost::bind(&SessionManager::OnHttpAccept, this,
		boost::asio::placeholders::error,
		pNewSocket, pAcceptor));
}

void SessionManager::RealCloseSession(SocketID sid)
{
	std::map<SocketID, boost::shared_ptr<TCPSession> >::iterator iter = m_SessionMap.find(sid);
	if (iter != m_SessionMap.end())
	{
		m_SessionMap.erase(iter);
	}
}

void SessionManager::Close(SocketID sid)
{
	if (m_SessionMap.empty())
	{
		return;
	}

	m_io.dispatch(boost::bind(&SessionManager::RealCloseSession, this, sid));
}

void SessionManager::RealStop()
{
	for (size_t n = 0; n < m_acceptors.size(); ++n)
	{
		m_acceptors[n]->close();
	}
	m_acceptors.clear();

	std::map<SocketID, boost::shared_ptr<TCPSession> > tempSessions = m_SessionMap;
	m_SessionMap.clear();

	for (std::map<SocketID, boost::shared_ptr<TCPSession> >::iterator iter = tempSessions.begin();
		iter != tempSessions.end(); ++iter)
	{
		iter->second->Close();
	}
	tempSessions.clear();

	m_io.stop();
	m_io_sockets.wait_for_stop();
}

SendResult SessionManager::AsyncSend(SocketID nSID, const char *pBuffer, int size, CONTENT_TYPE type)
{
	boost::recursive_mutex::scoped_lock	lock(m_recursive_mutex);

	std::map<SocketID, boost::shared_ptr<TCPSession> >::iterator iter = m_SessionMap.find(nSID);
	if (iter != m_SessionMap.end())
	{
		return iter->second->AsyncSend(pBuffer, size, type);
	}

	return RESULT_SESSION_CLOSED;
}

SendResult SessionManager::SyncSend(SocketID nSID, const char *pBuffer, int size, CONTENT_TYPE type)
{
	boost::recursive_mutex::scoped_lock	lock(m_recursive_mutex);

	std::map<SocketID, boost::shared_ptr<TCPSession> >::iterator iter = m_SessionMap.find(nSID);
	if (iter != m_SessionMap.end())
	{
		return iter->second->SyncSend(pBuffer, size, type);
	}

	return RESULT_OK;
}

void SessionManager::BoradcastTo(const std::vector<SocketID> ids, const char *pBuffer, int size, CONTENT_TYPE type)
{
	
	for (size_t n = 0; n < ids.size(); ++n)
	{
		boost::shared_ptr<TCPSession> pSession;
		{
			boost::recursive_mutex::scoped_lock	lock(m_recursive_mutex);
			std::map<SocketID, boost::shared_ptr<TCPSession> >::iterator iter = m_SessionMap.find(ids[n]);
			if (iter != m_SessionMap.end())
			{
				pSession = iter->second;
			}
		}

		if (pSession)
		{
			pSession->AsyncSend(pBuffer, size, type);
		}
	}
}

void SessionManager::BoradcastToAll(const char *pBuffer, int size, CONTENT_TYPE type)
{
	std::map<SocketID, boost::shared_ptr<TCPSession> > tempSessions;
	{
		boost::recursive_mutex::scoped_lock	lock(m_recursive_mutex);
		tempSessions = m_SessionMap;
	}

	std::map<SocketID, boost::shared_ptr<TCPSession> >::iterator iter = tempSessions.begin();
	for (; iter != tempSessions.end(); ++iter)
	{
		boost::shared_ptr<TCPSession> pSession = iter->second;
		if (pSession)
		{
			pSession->AsyncSend(pBuffer, size, type);
		}
	}
}

boost::asio::io_service& SessionManager::get_io_service()
{
	return  *m_io_sockets.get_io_service();
}

void SessionManager::check_socket(int expired_max)
{
	std::map<SocketID, boost::shared_ptr<TCPSession> > tempSessions;
	{
		boost::recursive_mutex::scoped_lock	lock(m_recursive_mutex);
		tempSessions = m_SessionMap;
	}

	time_t tNow = time(NULL);

	std::map<SocketID, boost::shared_ptr<TCPSession> >::iterator iter = tempSessions.begin();
	for (; iter != tempSessions.end(); ++iter)
	{
		boost::shared_ptr<TCPSession> pSession = iter->second;
		if (tNow - pSession->GetLastRecvTime() >= expired_max)
		{
			pSession->Close();
		}
	}
}
